package net.jgrm.redisadvanced.bloomfilter;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import net.jgrm.redisadvanced.config.RedisAdvancedProperties;
import org.springframework.data.redis.core.RedisTemplate;

import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Redis布隆过滤器服务实现类
 * 使用Redis的位图功能实现布隆过滤器
 */
@Slf4j
//@Service
public class RedisBloomFilterServiceImpl implements RedisBloomFilterService {

    //    @Autowired
    private RedisTemplate<String, Object> redisTemplate;

    //    @Autowired
    private RedisAdvancedProperties redisAdvancedProperties;

    //    @Autowired
    private ObjectMapper objectMapper;

    public RedisBloomFilterServiceImpl(RedisTemplate<String, Object> redisTemplate, RedisAdvancedProperties redisAdvancedProperties, ObjectMapper objectMapper) {
        this.redisTemplate = redisTemplate;
        this.redisAdvancedProperties = redisAdvancedProperties;
        this.objectMapper = objectMapper;
    }

    // 存储布隆过滤器配置的缓存
    private final Map<String, BloomFilterConfig> filterConfigCache = new ConcurrentHashMap<>();

    // 布隆过滤器配置的键前缀
    private static final String CONFIG_PREFIX = "bf:config:";
    // 布隆过滤器位图的键前缀
    private static final String BITMAP_PREFIX = "bf:bitmap:";

    /**
     * 添加前缀到键名
     */
    private String addPrefix(String key) {
        return redisAdvancedProperties.getKeyPrefix() + key;
    }

    @Override
    public void createFilter(String name, long expectedElements, double falsePositiveRate) {
        try {
            log.debug("Creating bloom filter: name={}, expectedElements={}, falsePositiveRate={}",
                    name, expectedElements, falsePositiveRate);

            // 验证参数
            if (expectedElements <= 0) {
                throw new IllegalArgumentException("Expected elements must be positive");
            }
            if (falsePositiveRate <= 0 || falsePositiveRate >= 1) {
                throw new IllegalArgumentException("False positive rate must be between 0 and 1");
            }

            // 计算布隆过滤器的参数
            long bitSize = calculateBitSize(expectedElements, falsePositiveRate);
            int numHashFunctions = calculateOptimalHashFunctions(bitSize, expectedElements);

            // 创建配置对象
            BloomFilterConfig config = new BloomFilterConfig(name, expectedElements, falsePositiveRate, numHashFunctions, bitSize);

            // 保存配置到Redis
            String configKey = addPrefix(CONFIG_PREFIX + name);
            redisTemplate.opsForValue().set(configKey, objectMapper.writeValueAsString(config));

            // 缓存配置
            filterConfigCache.put(name, config);

            log.info("Bloom filter created successfully: {}", config);
        } catch (Exception e) {
            log.error("Failed to create bloom filter: name={}, error={}", name, e.getMessage(), e);
            throw new RuntimeException("Failed to create bloom filter", e);
        }
    }

    @Override
    public boolean add(String name, Object element) {
        try {
            // 获取布隆过滤器配置
            BloomFilterConfig config = getFilterConfig(name);
            if (config == null) {
                throw new IllegalStateException("Bloom filter not found: " + name);
            }

            // 计算元素的哈希值并设置到位图
            String bitmapKey = addPrefix(BITMAP_PREFIX + name);
            boolean anySet = false;

            byte[] elementBytes = serializeElement(element);
            for (int i = 0; i < config.getNumHashFunctions(); i++) {
                long bitIndex = getHashIndex(elementBytes, i, config.getBitSize());
                Boolean previousValue = redisTemplate.opsForValue().setBit(bitmapKey, bitIndex, true);
                if (!previousValue) {
                    anySet = true;
                }
            }

            log.debug("Added element to bloom filter: name={}", name);
            return anySet;
        } catch (Exception e) {
            log.error("Failed to add element to bloom filter: name={}, error={}", name, e.getMessage(), e);
            throw new RuntimeException("Failed to add element to bloom filter", e);
        }
    }

    @Override
    public long addAll(String name, Collection<Object> elements) {
        try {
            // 获取布隆过滤器配置
            BloomFilterConfig config = getFilterConfig(name);
            if (config == null) {
                throw new IllegalStateException("Bloom filter not found: " + name);
            }

            String bitmapKey = addPrefix(BITMAP_PREFIX + name);
            long addedCount = 0;

            for (Object element : elements) {
                boolean elementAdded = false;
                byte[] elementBytes = serializeElement(element);

                for (int i = 0; i < config.getNumHashFunctions(); i++) {
                    long bitIndex = getHashIndex(elementBytes, i, config.getBitSize());
                    Boolean previousValue = redisTemplate.opsForValue().setBit(bitmapKey, bitIndex, true);
                    if (!previousValue) {
                        elementAdded = true;
                    }
                }

                if (elementAdded) {
                    addedCount++;
                }
            }

            log.debug("Added {} elements to bloom filter: name={}", addedCount, name);
            return addedCount;
        } catch (Exception e) {
            log.error("Failed to add elements to bloom filter: name={}, error={}", name, e.getMessage(), e);
            throw new RuntimeException("Failed to add elements to bloom filter", e);
        }
    }

    @Override
    public boolean mightContain(String name, Object element) {
        try {
            // 获取布隆过滤器配置
            BloomFilterConfig config = getFilterConfig(name);
            if (config == null) {
                throw new IllegalStateException("Bloom filter not found: " + name);
            }

            String bitmapKey = addPrefix(BITMAP_PREFIX + name);
            byte[] elementBytes = serializeElement(element);

            // 检查所有哈希位置
            for (int i = 0; i < config.getNumHashFunctions(); i++) {
                long bitIndex = getHashIndex(elementBytes, i, config.getBitSize());
                Boolean bitValue = redisTemplate.opsForValue().getBit(bitmapKey, bitIndex);
                if (bitValue == null || !bitValue) {
                    // 如果任何一个位为0，则元素一定不存在
                    return false;
                }
            }

            // 所有位都为1，元素可能存在
            log.debug("Element might contain in bloom filter: name={}", name);
            return true;
        } catch (Exception e) {
            log.error("Failed to check element in bloom filter: name={}, error={}", name, e.getMessage(), e);
            throw new RuntimeException("Failed to check element in bloom filter", e);
        }
    }

    @Override
    public Collection<Object> mightContainAll(String name, Collection<Object> elements) {
        try {
            List<Object> mightContainElements = new ArrayList<>();
            for (Object element : elements) {
                if (mightContain(name, element)) {
                    mightContainElements.add(element);
                }
            }
            return mightContainElements;
        } catch (Exception e) {
            log.error("Failed to check elements in bloom filter: name={}, error={}", name, e.getMessage(), e);
            throw new RuntimeException("Failed to check elements in bloom filter", e);
        }
    }

    @Override
    public boolean deleteFilter(String name) {
        try {
            String configKey = addPrefix(CONFIG_PREFIX + name);
            String bitmapKey = addPrefix(BITMAP_PREFIX + name);

            // 删除配置和位图
            redisTemplate.delete(Arrays.asList(configKey, bitmapKey));

            // 从缓存中移除配置
            filterConfigCache.remove(name);

            log.debug("Bloom filter deleted: name={}", name);
            return true;
        } catch (Exception e) {
            log.error("Failed to delete bloom filter: name={}, error={}", name, e.getMessage(), e);
            throw new RuntimeException("Failed to delete bloom filter", e);
        }
    }

    @Override
    public boolean exists(String name) {
        try {
            // 先检查缓存
            if (filterConfigCache.containsKey(name)) {
                return true;
            }

            // 检查Redis
            String configKey = addPrefix(CONFIG_PREFIX + name);
            Boolean exists = redisTemplate.hasKey(configKey);

            // 如果存在，加载到缓存
            if (exists != null && exists) {
                loadFilterConfig(name);
            }

            return exists != null && exists;
        } catch (Exception e) {
            log.error("Failed to check bloom filter existence: name={}, error={}", name, e.getMessage(), e);
            throw new RuntimeException("Failed to check bloom filter existence", e);
        }
    }

    @Override
    public BloomFilterConfig getConfig(String name) {
        return getFilterConfig(name);
    }

    /**
     * 获取布隆过滤器配置，从缓存或Redis中加载
     */
    private BloomFilterConfig getFilterConfig(String name) {
        // 先检查缓存
        BloomFilterConfig config = filterConfigCache.get(name);
        if (config == null) {
            // 缓存中没有，从Redis加载
            config = loadFilterConfig(name);
        }
        return config;
    }

    /**
     * 从Redis加载布隆过滤器配置
     */
    private BloomFilterConfig loadFilterConfig(String name) {
        try {
            String configKey = addPrefix(CONFIG_PREFIX + name);
            String configJson = (String) redisTemplate.opsForValue().get(configKey);
            if (configJson != null) {
                BloomFilterConfig config = objectMapper.readValue(configJson, BloomFilterConfig.class);
                filterConfigCache.put(name, config);
                return config;
            }
            return null;
        } catch (Exception e) {
            log.error("Failed to load bloom filter config: name={}, error={}", name, e.getMessage(), e);
            throw new RuntimeException("Failed to load bloom filter config", e);
        }
    }

    /**
     * 计算布隆过滤器的最佳位数
     * m = -n*ln(p)/(ln(2)^2)
     * 其中：m是位数，n是预期元素数，p是误判率
     */
    private long calculateBitSize(long expectedElements, double falsePositiveRate) {
        return (long) Math.ceil(-expectedElements * Math.log(falsePositiveRate) / (Math.log(2) * Math.log(2)));
    }

    /**
     * 计算布隆过滤器的最佳哈希函数数量
     * k = (m/n)*ln(2)
     * 其中：k是哈希函数数量，m是位数，n是预期元素数
     */
    private int calculateOptimalHashFunctions(long bitSize, long expectedElements) {
        return (int) Math.ceil((bitSize / expectedElements) * Math.log(2));
    }

    /**
     * 序列化元素为字节数组
     */
    private byte[] serializeElement(Object element) {
        try {
            if (element == null) {
                throw new IllegalArgumentException("Element cannot be null");
            }
            if (element instanceof String) {
                return ((String) element).getBytes(StandardCharsets.UTF_8);
            } else if (element instanceof byte[]) {
                return (byte[]) element;
            } else {
                return objectMapper.writeValueAsBytes(element);
            }
        } catch (Exception e) {
            log.error("Failed to serialize element: {}", e.getMessage(), e);
            throw new RuntimeException("Failed to serialize element", e);
        }
    }

    /**
     * 获取元素在指定位图大小下的哈希索引
     * 使用双重哈希技术生成多个哈希函数
     */
    private long getHashIndex(byte[] elementBytes, int hashFunctionIndex, long bitSize) {
        try {
            // 使用MurmurHash的思想，通过组合两个不同的哈希函数生成多个哈希值
            MessageDigest md5 = MessageDigest.getInstance("MD5");
            byte[] hashBytes = md5.digest(elementBytes);

            // 从MD5哈希中提取两个32位整数
            int h1 = ((hashBytes[0] & 0xFF) << 24) |
                    ((hashBytes[1] & 0xFF) << 16) |
                    ((hashBytes[2] & 0xFF) << 8) |
                    (hashBytes[3] & 0xFF);

            int h2 = ((hashBytes[4] & 0xFF) << 24) |
                    ((hashBytes[5] & 0xFF) << 16) |
                    ((hashBytes[6] & 0xFF) << 8) |
                    (hashBytes[7] & 0xFF);

            // 生成第k个哈希值：h1 + k*h2
            long combinedHash = (long) h1 + (long) hashFunctionIndex * (long) h2;
            // 确保结果为正
            if (combinedHash < 0) {
                combinedHash = -combinedHash;
            }
            // 取模得到最终的位索引
            return combinedHash % bitSize;
        } catch (NoSuchAlgorithmException e) {
            log.error("Failed to compute hash: {}", e.getMessage(), e);
            throw new RuntimeException("Failed to compute hash", e);
        }
    }
}